package com.jiepuxun.demo.flink;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

public class TestFlink{

    public void test() throws Exception{
        System.out.println("=========  流程开始   >>>>>>>>>  演示Flink Job  <<<<<<<<<<<<<<   ========== ");
        //1.准备环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //2.准备数据
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers","localhost:9092");
        properties.setProperty("group.id","flink");
        DataStreamSource<String> stream = env.addSource(
                new FlinkKafkaConsumer<String>("test", new SimpleStringSchema(), properties)
        );

        //4.设置flink 任务
        stream.map(new MapFunction<String, String>() {
            @Override
            public String map(String s) throws Exception {
                return "flink : " + s;
            }
        }).print();

        try {
            //5.触发执行
            env.execute();
        } catch (Exception e) {
            System.out.println("Error executing flink job: " + e.getMessage());
        }
        System.out.println("******演示结束******");
    }
}
